package com.ehualu.impda.hadoop;

import com.ehualu.impda.kafkasecurity.LoginUtilKafkaHuawei;
import com.ehualu.impda.security.LoginUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.File;
import java.io.IOException;
import java.util.List;

/**
 * @Author: 吴敬超
 * @Date: 2019/9/20 10:11
 */
public class Hdfsdemo {


    private static final String USER_KEYTAB_FILE = "user.keytab";

    /**
     * 用户自己申请的机机账号名称
     */
    private static final String USER_PRINCIPAL = "super_rj";


    private static final String STORAGE_POLICY_HOT = "HOT";
    private static String PATH_TO_HDFS_SITE_XML = HdfsExample2.class.getClassLoader().getResource("hdfs-site.xml")
            .getPath();
    private static String PATH_TO_CORE_SITE_XML = HdfsExample2.class.getClassLoader().getResource("core-site.xml")
            .getPath();

    private static Configuration conf = null;

    private static String PRNCIPAL_NAME = "super_rj";
    private static String PATH_TO_KEYTAB = HdfsExample2.class.getClassLoader().getResource("user.keytab").getPath();
    private static String PATH_TO_KRB5_CONF = HdfsExample2.class.getClassLoader().getResource("krb5.conf").getPath();

    /**
     * Add configuration file if the application run on the linux ,then need
     * make the path of the core-site.xml and hdfs-site.xml to in the linux
     * client file
     *
     * @return
     */
    private static void confLoad() throws IOException {
        System.setProperty("java.security.krb5.conf", PATH_TO_KRB5_CONF);
        conf = new Configuration();
        // conf file
        conf.addResource(new Path(PATH_TO_HDFS_SITE_XML));
        conf.addResource(new Path(PATH_TO_CORE_SITE_XML));

        // conf.addResource(new Path(PATH_TO_SMALL_SITE_XML));
    }

    /**
     * kerberos security authentication if the application running on Linux,need
     * the path of the krb5.conf and keytab to edit to absolute path in Linux.
     * make the keytab and principal in example to current user's keytab and
     * username
     */
    private static void authentication() throws IOException {
        // security mode
        if ("kerberos".equalsIgnoreCase(conf.get("hadoop.security.authentication"))) {
            System.setProperty("java.security.krb5.conf", PATH_TO_KRB5_CONF);
            LoginUtil.login(PRNCIPAL_NAME, PATH_TO_KEYTAB, PATH_TO_KRB5_CONF, conf);
        }
    }

    public static void securityPrepare() throws IOException {
        String filePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "resources" + File.separator;
        String krbFile = filePath + "krb5.conf";
        String userKeyTableFile = filePath + USER_KEYTAB_FILE;

        //windows路径下分隔符替换
        userKeyTableFile = userKeyTableFile.replace("\\", "\\\\");
        krbFile = krbFile.replace("\\", "\\\\");

        LoginUtilKafkaHuawei.setKrb5Config(krbFile);
        LoginUtilKafkaHuawei.setZookeeperServerPrincipal("zookeeper/hadoop.hadoop.com");
        LoginUtilKafkaHuawei.setJaasFile(USER_PRINCIPAL, userKeyTableFile);
    }


//    private void writeHDFS(String js) throws IOException {
//        final String content = js;
//        FSDataOutputStream out = null;
//        Path path = new Path(DEST_PATH + File.separator + FILE_NAME);
//        try {
//            boolean fiex = fSystem.exists(path);
//
//            if (fiex) {
//
//                System.out.println("文件已经存在");
//                out = fSystem.append(path);
//                out.write(content.getBytes());
//                out.hsync();
//                LOG.info("append to write.");
//            } else {
//
//                System.out.println("文件不存在");
//
//
//                out = fSystem.create(path);
//                out.write(content.getBytes());
//                out.hsync();
//                LOG.info("success to write.");
//            }
//
//        } finally {
//            // make sure the stream is closed finally.
//            IOUtils.closeStream(out);
//        }
//    }


    private void writeHDFS3(String js, FileSystem fSystem, FSDataOutputStream out, Path path) throws IOException {
        final String content = js;
//        FSDataOutputStream out = null;
//        Path path = new Path(DEST_PATH + File.separator + FILE_NAME);
        try {
            boolean fiex = fSystem.exists(path);

            if (fiex) {

                System.out.println("文件已经存在");
                out = fSystem.append(path);
                out.write(content.getBytes());
                out.hsync();
                System.out.println("append to write.");
            } else {

                System.out.println("文件不存在");


                out = fSystem.create(path);
                out.write(content.getBytes());
                out.hsync();
                System.out.println("success to write.");
            }

        } finally {
            System.out.println("***********");
            // make sure the stream is closed finally.
//            IOUtils.closeStream(out);
        }
    }


    public static void main(String[] args) throws Exception {


//        FileSystem fSystem = null;


        confLoad();

        authentication();
        securityPrepare();
//        String hf="test889.txt";
        FileSystem fSystem = FileSystem.get(conf);
        FSDataOutputStream out = null;
        String DEST_PATH = "/user/hdfs-examples";
        String FILE_NAME = "test889.txt";


        String content = "write to hdfs ";

        Path path = new Path(DEST_PATH + File.separator + FILE_NAME);


        for (int num = 0; num < 3; num++) {

            System.out.println("*******循环写入次数*****:" + num);
//            hdfs_examples.writeHDFS(result);


            boolean fiex = fSystem.exists(path);

            System.out.println(fiex);


            if (fiex) {

                System.out.println("文件已经存在");
                out = fSystem.append(path);
                out.write(content.getBytes());
                out.hsync();

                System.out.println("append to write.");
//            LOG.info("append to write.");
            } else {

                System.out.println("文件不存在");


                out = fSystem.create(path);
                out.write(content.getBytes());
                out.hsync();
                System.out.println("success to write.");
//            LOG.info("success to write.");
            }

            IOUtils.closeStream(out);
        }



    }
}
